package com.atguigu.flink.sqlfunction;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/6
 */
public class Demo10_IntervalJoin
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());

        //必须是keyedStream
        KeyedStream<WaterSensor, String> ds1 = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(WaterSensor::getId);

        KeyedStream<WaterSensor, String> ds2 = env
            .socketTextStream("hadoop103", 8889)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(WaterSensor::getId);
        /*
                开启基于事件时间的窗口

                ds1 join ds2 on ds1.xx = ds2.xx

                全程没有开窗。
         */
        ds1.intervalJoin(ds2)
           //目前已经过时了，禁止使用，用了就报错
           //.inProcessingTime()
           //要join的对方流的时间范围
           .between(Time.seconds(-3),Time.seconds(3))
           .process(new ProcessJoinFunction<WaterSensor, WaterSensor, String>()
           {
               @Override
               public void processElement(WaterSensor left, WaterSensor right, Context ctx, Collector<String> out) throws Exception {
                   out.collect(left + "====" + right);
               }
           })
           .print();
           //如果数据在边界线上达到，默认关联，可以设置不关联
           //.upperBoundExclusive()
           //.lowerBoundExclusive()
           //默认
           //.inEventTime()



        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
